package org.lld.im.service.message.service;

import com.alibaba.fastjson.JSONObject;
import com.lld.im.codec.pack.message.ChatMessageAck;
import com.lld.im.codec.pack.message.MessageReciveServerAckPack;
import org.lld.im.common.ResponseVO;
import org.lld.im.common.config.AppConfig;
import org.lld.im.common.constant.Constants;
import org.lld.im.common.enums.ConversationTypeEnum;
import org.lld.im.common.enums.command.MessageCommand;
import org.lld.im.common.model.ClientInfo;
import org.lld.im.common.model.message.MessageContent;
import org.lld.im.common.model.message.OfflineMessageContent;
import org.lld.im.service.message.model.req.SendMessageReq;
import org.lld.im.service.message.model.resp.SendMessageResp;
import org.lld.im.service.seq.RedisSeq;
import org.lld.im.service.utils.CallbackService;
import org.lld.im.service.utils.ConversationIdGenerate;
import org.lld.im.service.utils.MessageProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeanUtils;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author: LunBoWang
 * @description: 单聊消费者，用来接收im发给逻辑层的消息
 * @date: 2025/4/3 下午5:33
 * @version: 1.0
 */
@Service
public class P2PMessageService {

    private static Logger logger = LoggerFactory.getLogger(P2PMessageService.class);

    @Resource
    CheckSendMessageService checkSendMessageService;

    @Resource
    MessageProducer messageProducer;

    @Resource
    MessageStoreService messageStoreService;

    @Resource
    RedisSeq redisSeq;

    @Resource
    AppConfig appConfig;

    @Resource
    CallbackService callbackService;

    private final ThreadPoolExecutor threadPoolExecutor;

    {
        final AtomicInteger num = new AtomicInteger(0);
        threadPoolExecutor = new ThreadPoolExecutor(8, 8, 60, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(1000), new ThreadFactory() {
            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setDaemon(true);
                thread.setName("message-process-thread-" + num.getAndIncrement());
                return thread;
            }
        });
    }

    public void process(MessageContent messageContent) {
        String fromId = messageContent.getFromId();
        String toId = messageContent.getToId();
        Integer appId = messageContent.getAppId();
        //用messageId从缓存中获取消息。
        MessageContent messageFromMessageIdCache = messageStoreService.getMessageFromMessageIdCache
                (messageContent.getAppId(), messageContent.getMessageId(), MessageContent.class);
        if (messageFromMessageIdCache != null) {
            threadPoolExecutor.execute(() -> {
                //1.回ack成功给自己
                ack(messageContent, ResponseVO.successResponse());
                //2.发消息给同步在线端
                syncToSender(messageFromMessageIdCache, messageFromMessageIdCache);
                //3.发消息给对方在线端
                List<ClientInfo> clientInfos = dispatchMessage(messageFromMessageIdCache);
                if (clientInfos.isEmpty()) {
                    //发送接收确认给发送方，要带上是服务端发送的标识
                    reviceAck(messageContent);
                }
            });
            return;
        }

        //回调
        ResponseVO responseVO = ResponseVO.successResponse();
        if(appConfig.isSendMessageBeforeCallback()){
            responseVO=callbackService.beforeCallback(messageContent.getAppId(), Constants.CallbackCommand.SendMessageBefore
            , JSONObject.toJSONString(messageContent));
        }

        if(!responseVO.isOk()){
            //告诉客户端失败了
            //ack
            ack(messageContent, responseVO);
            return;
        }

        // 前置校验
        //这个用户是否被禁言 是否被禁用
        //发送方和接收方是否是好友关系
//        ResponseVO responseVO = imServerPermissionCheck(fromId, toId, appId);
//        if (responseVO.isOk()) {
        long seq = redisSeq.doGetSeq(messageContent.getAppId()
                + ":" + Constants.SeqConstants.Message + ":" + ConversationIdGenerate.generateP2PId(
                messageContent.getFromId(), messageContent.getToId()
        ));
        messageContent.setMessageSequence(seq);

        threadPoolExecutor.execute(() -> {
            //appId + Seq + (from+to) froupId
            messageStoreService.storeP2pMessage(messageContent);

            //插入离线消息
            OfflineMessageContent offlineMessageContent = new OfflineMessageContent();
            BeanUtils.copyProperties(messageContent, offlineMessageContent);
            offlineMessageContent.setConversationType(ConversationTypeEnum.P2P.getCode());
            messageStoreService.storeOfflineMessage(offlineMessageContent);

            //插入数据
            //1.回ack成功给自己
            ack(messageContent, ResponseVO.successResponse());
            //2.发消息给同步在线端
            syncToSender(messageContent, messageContent);
            //3.发消息给对方在线端
            List<ClientInfo> clientInfos = dispatchMessage(messageContent);
            //将messageId存到缓存中
            messageStoreService.setMessageFromMessageIdCache(messageContent.getAppId(),
                    messageContent.getMessageId(), messageContent);

            if (clientInfos.isEmpty()) {
                //发送接收确认给发送方，要带上是服务端发送的标识
                reviceAck(messageContent);
            }
            if(appConfig.isSendMessageAfterCallback()){
                callbackService.callback(messageContent.getAppId(), Constants.CallbackCommand.SendMessageAfter
                        , JSONObject.toJSONString(messageContent));
            }
        });
//        } else {
//            //告诉客户端失败了
//            //ack
//            ack(messageContent, responseVO);
//        }
    }

    private List<ClientInfo> dispatchMessage(MessageContent messageContent) {
        return messageProducer.sendToUser(messageContent.getToId(), MessageCommand.MSG_P2P,
                messageContent, messageContent.getAppId());
    }

    private void ack(MessageContent messageContent, ResponseVO responseVO) {
        logger.info("msg ack,msgId={},checkResult={}", messageContent.getMessageId(), responseVO.getCode());

        ChatMessageAck chatMessageAck = new ChatMessageAck(messageContent.getMessageId(), messageContent.getMessageSequence());
        responseVO.setData(chatMessageAck);
        //发消息
        messageProducer.sendToUser(messageContent.getFromId(), MessageCommand.MSG_ACK,
                responseVO, messageContent);
    }

    private void reviceAck(MessageContent messageContent) {
        MessageReciveServerAckPack pack = new MessageReciveServerAckPack();
        pack.setFromId(messageContent.getToId());
        pack.setToId(messageContent.getFromId());
        pack.setMessageKey(messageContent.getMessageKey());
        pack.setMessageSquence(messageContent.getMessageSequence());
        pack.setServerSend(true);
        messageProducer.sendToUser(messageContent.getFromId(), MessageCommand.MSG_RECIVE_ACK,
                pack, new ClientInfo(messageContent.getAppId(), messageContent.getClientType(),
                        messageContent.getImei()));
    }

    private void syncToSender(MessageContent messageContent, ClientInfo clientInfo) {
        messageProducer.sendToUserExceptClient(messageContent.getFromId(),
                MessageCommand.MSG_P2P, messageContent, messageContent);
    }

    public ResponseVO imServerPermissionCheck(String fromId, String toId,
                                              Integer appId) {
        ResponseVO responseVO = checkSendMessageService.checkSenderForvidAndMute(fromId, appId);
        if (!responseVO.isOk()) {
            return responseVO;
        }
        responseVO = checkSendMessageService.checkFriendShip(fromId, toId, appId);
        // 权限校验
        return responseVO;
    }

    public SendMessageResp send(SendMessageReq req) {
        SendMessageResp sendMessageResp = new SendMessageResp();
        MessageContent messageContent = new MessageContent();
        BeanUtils.copyProperties(req, messageContent);
        //插入数据
        messageStoreService.storeP2pMessage(messageContent);
        sendMessageResp.setMessageKey(messageContent.getMessageKey());
        sendMessageResp.setMessageTime(System.currentTimeMillis());

        //发送消息给同步在线端
        syncToSender(messageContent, messageContent);
        //发送消息给对方在线端
        dispatchMessage(messageContent);
        return sendMessageResp;
    }
}
